#pragma once

#include "../communal/Communication.hpp"
#include "../communal/Message.hpp"
#include <unordered_set>

namespace xu
{
    namespace server
    {
        class TopicManager
        {
        public:
            using Ptr = std::shared_ptr<TopicManager>;
            TopicManager(){}

            void onTopicRequest(const BaseConnection::Ptr &conn, const TopicRequest::Ptr &msg)
            {
                TopicOptype optype =  msg->Optype();
                bool ret = true;
                switch (optype)
                {
                case TopicOptype::TOPIC_CREATE :TopicCreate(conn,msg); break;
                case TopicOptype::TOPIC_REMOVE :TopicRemove(conn,msg); break;
                case TopicOptype::TOPIC_SUBSCRIBE : ret = topicSubscribe(conn,msg); break;
                case TopicOptype::TOPIC_CANCEL :topicCancel(conn,msg); break;
                case TopicOptype::TOPIC_PUBLISH :ret = topicPublish(conn,msg); break;
                default: return  errorResponse(conn,msg,RCode::RCODE_INVALID_OPTYPE) ;
                }

                if(ret == false)
                {
                    return errorResponse(conn,msg,RCode::RCODE_NOT_FOUND_TOPIC);
                }

                return topicResponse(conn,msg);
            }
            void onShutdown(const BaseConnection::Ptr &conn)
            {
                //1.找到订阅者并从_subscribers移除
                Subscriber::Ptr subscriber;
                std::vector<Topic::Ptr> topics;
                {
                    std::unique_lock<std::mutex> _lock(_mutex);
                    auto it = _subscribers.find(conn);
                    if(it == _subscribers.end())
                    {
                        return;
                    }

                    subscriber = it->second;
                    _subscribers.erase(conn);

                    //2.找到订阅了哪些主题
                    for(auto &e : subscriber->topics)
                    {
                        auto it = _topics.find(e);
                        if(it != _topics.end())
                        {
                            topics.push_back(it->second);
                        }
                    }
                }

                //3.从主题中删除订阅者
                for(auto &e : topics)
                {
                    e->Remove(subscriber);
                }
            
            }

        private:
            //出错响应
            void errorResponse(const BaseConnection::Ptr &conn, const TopicRequest::Ptr &msg, RCode rcode)
            {
                auto rsp = MessageFactory::create<TopicResponse>();
                rsp->SetId(msg->Id());
                rsp->SetRcode(rcode);
                rsp->SetMType(MType::RSP_TOPIC);
                conn->Send(rsp);
            }

            //正确响应
            void topicResponse(const BaseConnection::Ptr &conn, const TopicRequest::Ptr &msg)
            {
                auto rsp = MessageFactory::create<TopicResponse>();
                rsp->SetId(msg->Id());
                rsp->SetRcode(RCode::RCODE_OK);
                rsp->SetMType(MType::RSP_TOPIC);
                conn->Send(rsp);
            }

            // 创建主题
            void TopicCreate(const BaseConnection::Ptr &conn, const TopicRequest::Ptr &msg)
            {
                std::unique_lock<std::mutex> _lock(_mutex);
                auto topic = std::make_shared<Topic>(msg->Key());
                _topics.insert(std::make_pair(msg->Key(), topic));
            }

            // 删除主题
            void TopicRemove(const BaseConnection::Ptr &conn, const TopicRequest::Ptr &msg)
            {
                // 先找到主题
                std::unordered_set<Subscriber::Ptr> subscribers;
                std::string key = msg->Key();
                {
                    std::unique_lock<std::mutex> _lock(_mutex);
                    auto it = _topics.find(key);
                    if (it == _topics.end())
                    {
                        return;
                    }
                    subscribers = it->second->subscribers;
                    _topics.erase(key);
                }

                // 从订阅者中删除该主题
                for (auto &e : subscribers)
                {
                    e->Remove(key);
                }
            }

            // 订阅主题
            bool topicSubscribe(const BaseConnection::Ptr &conn, const TopicRequest::Ptr &msg)
            {
                // 先找到主题
                std::string key = msg->Key();
                Subscriber::Ptr subscriber;
                Topic::Ptr topic;
                {
                    std::unique_lock<std::mutex> _lock(_mutex);
                    auto it_topic = _topics.find(key);
                    if (it_topic == _topics.end()) // 不存在该主题报错
                    {
                        return false;
                    }

                    // 找该订阅者
                    auto it_sub = _subscribers.find(conn);
                    if (it_sub == _subscribers.end()) // 不存在则创建
                    {
                        subscriber = std::make_shared<Subscriber>(conn);
                        _subscribers.insert(std::make_pair(conn, subscriber));
                    }
                    else
                    {
                        subscriber = it_sub->second;
                    }

                    topic = it_topic->second;
                }

                // 主题插入该订阅者
                topic->Append(subscriber);

                // 该订阅者添加该主题
                subscriber->Append(key);

                return true;
            }

            // 取消订阅
            void topicCancel(const BaseConnection::Ptr &conn, const TopicRequest::Ptr &msg)
            {
                // 先找到主题
                std::string key = msg->Key();
                Subscriber::Ptr subscriber;
                Topic::Ptr topic;
                {
                    std::unique_lock<std::mutex> _lock(_mutex);
                    auto it_topic = _topics.find(key);
                    if (it_topic == _topics.end()) // 不存在该主题返回
                    {
                        return;
                    }

                    // 找该订阅者
                    auto it_sub = _subscribers.find(conn);
                    if (it_sub == _subscribers.end()) // 不存在则返回
                    {
                        return;
                    }

                    subscriber = it_sub->second;
                    topic = it_topic->second;
                }

                topic->Remove(subscriber);
                subscriber->Remove(key);
            }

            // 发布消息
            bool topicPublish(const BaseConnection::Ptr &conn, const TopicRequest::Ptr &msg)
            {
                Topic::Ptr topic;
                std::string key = msg->Key();
                {
                    std::unique_lock<std::mutex> _lock(_mutex);
                    auto it_topic = _topics.find(key);
                    if (it_topic == _topics.end()) // 不存在该主题报错
                    {
                        return false;
                    }

                    topic = it_topic->second;
                }

                topic->pushMessage(msg);

                return true;
            }

            // 订阅者
            struct Subscriber
            {
                using Ptr = std::shared_ptr<Subscriber>;
                std::mutex _mutex;
                BaseConnection::Ptr conn;
                std::unordered_set<std::string> topics;

                Subscriber(const BaseConnection::Ptr &c) : conn(c) {}
                void Append(const std::string &topic)
                {
                    std::unique_lock<std::mutex> _lock(_mutex);
                    topics.insert(topic);
                }
                void Remove(const std::string &topic)
                {
                    std::unique_lock<std::mutex> _lock(_mutex);
                    topics.erase(topic);
                }
            };
            // 主题
            struct Topic
            {
                using Ptr = std::shared_ptr<Topic>;
                std::mutex _mutex;
                std::string topic;
                std::unordered_set<Subscriber::Ptr> subscribers;
                Topic(const std::string &key) : topic(key) {}
                void Append(const Subscriber::Ptr &subscriber)
                {
                    std::unique_lock<std::mutex> _lock(_mutex);
                    subscribers.insert(subscriber);
                }
                void Remove(const Subscriber::Ptr &subscriber)
                {
                    std::unique_lock<std::mutex> _lock(_mutex);
                    subscribers.erase(subscriber);
                }
                void pushMessage(const BaseMessage::Ptr &msg)
                {
                    std::unique_lock<std::mutex> _lock(_mutex);
                    for (auto &e : subscribers)
                    {
                        if (e->conn != nullptr)
                        {
                            e->conn->Send(msg);
                        }
                    }
                }
            };

        private:
            std::mutex _mutex;
            std::unordered_map<std::string, Topic::Ptr> _topics;
            std::unordered_map<BaseConnection::Ptr, Subscriber::Ptr> _subscribers;
        };

    }

}